package com.education.realtime.app.dwd.db;


import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// 互动域 用户章节评价
public class DwdInteractionComment extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdInteractionComment().init(3015,"DwdInteractionComment");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"DwdInteractionComment");
        Table commentTable = tEnv.sqlQuery("select " +
                " data['id'] id," +
                " data['user_id'] user_id," +
                " data['chapter_id'] chapter_id," +
                " data['course_id'] course_id," +
                " data['comment_txt'] comment_txt," +
                " date_format(data['create_time'],'yyyy-MM-dd') date_id," +
                " data['create_time'] create_time," +
                " ts" +
                " from ods_db" +
                " where `table` = 'comment_info'" +
                " and `type` = 'insert'");
        tEnv.createTemporaryView("comment_table",commentTable);


        Table chapterTable = tEnv.sqlQuery("select " +
                " data['id'] id," +
                " data['chapter_name'] chapter_name" +
                " from ods_db" +
                " where `table` = 'chapter_info'" +
                " and `type` = 'bootstrap-insert' "
        );
        tEnv.createTemporaryView("chapter_table",chapterTable);

        Table resultTable = tEnv.sqlQuery("select " +
                "  comm.id," +
                "  comm.user_id," +
                "  comm.chapter_id," +
                "  chapter.chapter_name," +
                "  comm.course_id," +
                "  comm.comment_txt," +
                "  comm.date_id," +
                "  comm.create_time," +
                "  comm.ts" +
                " from comment_table comm" +
                " join chapter_table chapter" +
                " on comm.chapter_id = chapter.id");

        tEnv.executeSql("create table "+ Constant.DWD_INTERACTION_COMMENT+"(" +
                " id string," +
                " user_id string," +
                " chapter_id string," +
                " chapter_name string," +
                " course_id string," +
                " comment_txt string," +
                " date_id string," +
                " create_time string," +
                " ts string)" + SqlUtil.getKafkaWriteDDL(Constant.DWD_INTERACTION_COMMENT));
        resultTable.executeInsert(Constant.DWD_INTERACTION_COMMENT);
    }
}
